package git.soulbgm.handle;

import cn.hutool.core.thread.ThreadUtil;
import git.soulbgm.pojo.SendData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

import static git.soulbgm.common.Constants.TOPIC_SEND_DATA;
import static git.soulbgm.common.Constants.TOPIC_SEND_DATA_PREFIX;

/**
 * 发送数据处理
 *
 * @author SoulBGM
 * @date 2024-11-10
 */
@Component
@KafkaListener(topics = TOPIC_SEND_DATA)
// @KafkaListener(topicPattern = TOPIC_SEND_DATA_PREFIX + "[1-9]")
// @KafkaListener(topics = {TOPIC_SEND_DATA_PREFIX + "1", TOPIC_SEND_DATA_PREFIX + "2", TOPIC_SEND_DATA_PREFIX + "3", TOPIC_SEND_DATA_PREFIX + "4", TOPIC_SEND_DATA_PREFIX + "5", TOPIC_SEND_DATA_PREFIX + "6", TOPIC_SEND_DATA_PREFIX + "7", TOPIC_SEND_DATA_PREFIX + "8", TOPIC_SEND_DATA_PREFIX + "9"})
//, concurrency = "9"
public class SendDataHandle implements Runnable {

    private static final PriorityBlockingQueue<SendData> SEND_DATA_QUEUE = new PriorityBlockingQueue<>();

    public SendDataHandle() {
        CustomizableThreadFactory factory = new CustomizableThreadFactory();
        factory.setThreadNamePrefix("send-data-");
        RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
        int threadNum = 5;
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(threadNum, factory, handler);
        //ExecutorService service = ThreadUtil.newFixedExecutor(threadNum, 1000, "send-data-", handler);
        Runnable runnable = () -> {
            while (true) {
                SendData sendData = SEND_DATA_QUEUE.poll();
                if (sendData == null) {
                    break;
                }
                LOG.info("priority: {}, data: {}, time: {}", sendData.getPriority(), sendData.getData(), sendData.getTime());
            }
        };
        executor.scheduleAtFixedRate(runnable, 1, 1, TimeUnit.MILLISECONDS);
        /*for (int i = 0; i < threadNum; i++) {
            service.execute(this);
        }*/
    }

    private static final Logger LOG = LoggerFactory.getLogger(SendDataHandle.class);

    @KafkaHandler
    public void listen(@Header(KafkaHeaders.RECEIVED_TOPIC) String topic, String value) {
        SendData sendData = new SendData(topic, value);
        SEND_DATA_QUEUE.add(sendData);
    }

    @Override
    public void run() {
        while (true) {
            SendData sendData = SEND_DATA_QUEUE.poll();
            if (sendData == null) {
                continue;
            }
            LOG.info("priority: {}, data: {}, time: {}", sendData.getPriority(), sendData.getData(), sendData.getTime());
        }
    }
}
